/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package com.google.code.peersim.pastry.protocol;

import com.google.code.peersim.metrics.ProximityMetricIfc;
import com.google.code.peersim.pastry.protocol.PastryResourceAssignLsnrIfc.ResourceAssignedInfo;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import peersim.config.Configuration;
import peersim.core.CommonState;
import peersim.core.Fallible;
import peersim.core.GeneralNode;
import peersim.util.FileNameGenerator;

/**
 * Class representing a Pastry node. Adds useful methods to the standard PeerSim class
 * used for representing nodes.
 *
 * @author frusso
 * @version 0.1
 * @since 0.1
 */
public class PastryNode extends GeneralNode implements PastryProtocolListenerIfc {
  /**
   * This node's pastry protocol identifier.
   */
  public static final String PASTRY = "pastry";
  /**
   * Configures the name of the file to log to.
   */
  public static final String LOG_FILE = "log";
  /**
   * Whether nodes should log or not.
   */
  public static final String DO_LOG = "doLog";
  /**
   * Whether nodes should log or not.
   */
  public static final String MAX_LOOKUP_RETRIES = "maxLookupRetries";
  private int maxLookupRetries;
  /**
   * The protocol identifier associated to the Pastry protocol instance by the PeerSim runtime.
   */
  private final int pastryPID;
  /**
   * Whether to log node-activity or not.
   */
  private final boolean doLog;

  /**
   * The Pastry local-store for storing assigned pastry resources.
   */
  private Map<PastryId,PastryResource> localStore;
  /**
   * Instances registered to receive join-event notifications.
   */
  private Set<PastryJoinLsnrIfc> joinLsnrs;
  /**
   * Instances registered to receive resource-discovery event notifications.
   */
  private Set<PastryResourceDiscoveryLsnrIfc> resLookupLsnrs;
  /**
   * Instances registered to receive resource-assignment event notifications,
   * that is the assigners willing to know when they have been able to assign
   * a resource to a node.
   */
  private Set<PastryResourceAssignLsnrIfc> resAssignLsnrs;
  /**
   * Instances registered to receive resource-reception event notifications,
   * that is the assignees willing to know when they have been appointed to store
   * a resource.
   */
  private Set<PastryResourceAssignLsnrIfc> resRcvdLsnrs;
  /**
   * Instances registered to receive resource-routing event notifications.
   */
  private Set<PastryResourceAssignLsnrIfc> resRoutedLsnrs;
  /**
   * The file to log to.
   */
  private String fileName;
  /**
   * The stream to print to.
   */
  private PrintStream stream;
  private Queue<DelayedLookup> delayedLookups = new LinkedList<DelayedLookup>();
  private int delayedLookupTime;
  public static final String DELAYED_LOOKUP_TIME = "delayedLookupTime";
  private int unsentChunkReqDueToTimeout;

  /**
   * Constructor.
   * @param prefix
   */
  public PastryNode(String prefix) throws FileNotFoundException {
    super(prefix);
    pastryPID = Configuration.getPid(prefix+"."+PASTRY);
    getPastryProtocol().registerListener(this);
    doLog = Configuration.getBoolean(prefix + "." + DO_LOG);
    maxLookupRetries = Configuration.getInt(prefix + "." + MAX_LOOKUP_RETRIES);
    delayedLookupTime = Configuration.getInt(prefix + "." + DELAYED_LOOKUP_TIME);
    if(doLog) {
      fileName = new FileNameGenerator(Configuration.getString(prefix + "."+LOG_FILE), ".log").nextCounterName();
      stream = new PrintStream(new FileOutputStream(fileName));
    }
  }

  /**
   * Adds a listener for join events.
   *
   * @param lsnr The listener
   */
  public void addJoinListener(PastryJoinLsnrIfc lsnr) {
    if(joinLsnrs==null)
      joinLsnrs = new HashSet<PastryJoinLsnrIfc>();
    joinLsnrs.add(lsnr);
  }

  /**
   * Adds a listener for resource assignment events.
   *
   * @param lsnr The listener
   */
  public void addResourceAssignListener(PastryResourceAssignLsnrIfc lsnr) {
    if(resAssignLsnrs==null)
      resAssignLsnrs = new HashSet<PastryResourceAssignLsnrIfc>();
    resAssignLsnrs.add(lsnr);
  }

  /**
   * Adds a listener for resource reception events.
   *
   * @param lsnr The listener
   */
  public void addResourceReceivedListener(PastryResourceAssignLsnrIfc lsnr) {
    if(resRcvdLsnrs==null)
      resRcvdLsnrs = new HashSet<PastryResourceAssignLsnrIfc>();
    resRcvdLsnrs.add(lsnr);
  }

  /**
   * Adds a listener for resource routing events.
   *
   * @param lsnr The listener
   */
  public void addResourceRoutedListener(PastryResourceAssignLsnrIfc lsnr) {
    if(resRoutedLsnrs==null)
      resRoutedLsnrs = new HashSet<PastryResourceAssignLsnrIfc>();
    resRoutedLsnrs.add(lsnr);
  }

  /**
   * Adds a listener for resource discovery events.
   *
   * @param lsnr The listener
   */
  public void addResourceDiscoveryListener(PastryResourceDiscoveryLsnrIfc lsnr) {
    if(resLookupLsnrs==null)
      resLookupLsnrs = new HashSet<PastryResourceDiscoveryLsnrIfc>();
    resLookupLsnrs.add(lsnr);
  }

  /**
   * Tells this node to start checking for expired messages.
   */
  public void checkTimeouts() {
    // Pastry timeouts...
    this.getPastryProtocol().checkTimeouts();

    long currentTime = CommonState.getTime();
    // ... and node-related timeouts
    DelayedLookup[] pendings = delayedLookups.toArray(new DelayedLookup[delayedLookups.size()]);
    for(DelayedLookup pending : pendings) {
      // process the delayed lookup iff it's its resubmit time
      if(pending.resubmitTime == currentTime) {
        delayedLookups.remove(pending);
        // if we do not exceede the maximum number of retries
        if(++pending.retries<maxLookupRetries) {
          // we start the search again
          lookupResource(pending);
        } else {
          log("[NOT SENT] request "+pending+" not submitted due to max-retries reached");
          unsentChunkReqDueToTimeout++;
        }
      } else {
        // nop
      }
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Object clone() {
    // NOTE: this clone method is purposely not taking into account the sets of
    // listeneres registered on the current instance since it would be errouneous
    // copying them from this to the clone
    PastryNode clone = null;
    clone = (PastryNode) super.clone();
    // clone clean-up
    clone.joinLsnrs = null;
    clone.localStore = new HashMap<PastryId, PastryResource>();
    clone.resAssignLsnrs = null;
    clone.resLookupLsnrs = null;
    clone.resRcvdLsnrs = null;
    clone.delayedLookups = new LinkedList<DelayedLookup>();
    clone.getPastryProtocol().registerListener(clone);
    return clone;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean equals(Object obj) {
    if(this==obj)
      return true;
    if(!(obj instanceof PastryNode))
      return false;
    PastryNode other = (PastryNode) obj;
    return getPastryId().equals(other.getPastryId());
  }

  /**
   * Returns the {@link PastryId} assigned to this node.
   * @return The {@link PastryId} assigned to this node
   */
  public PastryId getPastryId() {
    return getPastryProtocol().getPastryId();
  }

  /**
   * Returns the pastry protocol identifier generated by the PeerSim simulator.
   * @return The pastry protocol id
   */
  public int getPastryPID() {
    return pastryPID;
  }

  /**
   * Returns the reference to the proximity protocol used by the Pastry protocol.
   * @return The proximity protocol used by the Pastry protocol
   */
  public <T extends ProximityMetricIfc.MeasurableEntityIfc> T getProximityProtocol() {
    return (T) getPastryProtocol().getProximityProtocol(this);
  }

  /**
   * Returns the reference to the Pastry protocol.
   * @return The Pastry protocol
   */
  public PastryProtocol getPastryProtocol() {
    PastryProtocol pp = null;
    pp = (PastryProtocol) this.getProtocol(pastryPID);
    return pp;
  }

  public PastryResource<?> getResource(PastryId resId) {
    return localStore.get(resId);
  }

  public int getUnsentChunkReqDueToTimeout() {
    return unsentChunkReqDueToTimeout;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public int hashCode() {
    return getPastryId().hashCode();
  }

  /**
   * Initializes this node.
   */
  public void initPastry() {
    getPastryProtocol().init(this);
  }

  /**
   * A node is joinable iff it is active and in state {@link PastryProtocolState#INITIALIZED}.
   * @return
   */
  public boolean isJoinable() {
    return super.isUp() && getPastryProtocol().isJoinable();
  }

  /**
   * A node is joined iff it is active and in state {@link PastryProtocolState#JOINED}.
   * @return
   */
  public boolean isJoined() {
    return super.isUp() && getPastryProtocol().isJoined();
  }

  /**
   * Tells whether the node has been appointed with a resource discovery task or not.
   *
   * @return Whether the node has been appointed with a resource discovery task or not
   */
  public boolean isSearching() {
    return getPastryProtocol().isSearching();
  }

  /**
   * Tells whether the node has been appointed with a resource diffusion task or not.
   *
   * @return Whether the node has been appointed with a resource diffusion task or not
   */
  public boolean isSending() {
    return getPastryProtocol().isSending();
  }

  /**
   * Tells the node to look for a resource whose id is the given one.
   * @param id The resource id to look for
   */
  public void lookupResource(PastryId id) {
    lookupResource(new DelayedLookup(id));
  }

  private void lookupResource(DelayedLookup search) {
    PastryId id = search.id;
    PastryResource<?> res = localStore.get(id);
    if(res==null) {
      // the resource is not locally available
      if(!getPastryProtocol().lookupResource(id)) {
        // but the search could not be started (maybe the state-tables are stale)
        delayedLookups.add(search);
      } else {
        // nop
      }
    } else {
      resourceDiscovered(new ResourceDiscoveredInfo(res, id, OK, this));
    }
  }

  /**
   * Tells the node to publish the given resource.
   * @param res The resource to be plubished
   */
  public void publishResource(PastryResource res) {
    getPastryProtocol().publishResource(res);
  }

  /**
   * Removes the given listener.
   * @param lsnr The listener
   */
  public void removeJoinListener(PastryJoinLsnrIfc lsnr) {
    if(joinLsnrs!=null)
      joinLsnrs.remove(lsnr);
  }

  /**
   * Removes the given listener.
   * @param lsnr The listener
   */
  public void removeResourceAssignListener(PastryResourceAssignLsnrIfc lsnr) {
    if(resAssignLsnrs!=null)
      resAssignLsnrs.remove(lsnr);
  }

  /**
   * Removes the specified listener for resource reception events.
   *
   * @param lsnr The listener
   */
  public void removeResourceReceivedListener(PastryResourceAssignLsnrIfc lsnr) {
    if(resRcvdLsnrs!=null)
      resRcvdLsnrs.remove(lsnr);
  }

  /**
   * Removes the specified listener for resource routing events.
   *
   * @param lsnr The listener
   */
  public void removeResourceRoutedListener(PastryResourceAssignLsnrIfc lsnr) {
    if(resRoutedLsnrs!=null)
      resRoutedLsnrs.remove(lsnr);
  }

  /**
   * Removes the given listener.
   * @param lsnr The listener
   */
  public void removeResourceLookupListener(PastryResourceDiscoveryLsnrIfc lsnr) {
    if(resLookupLsnrs!=null)
      resLookupLsnrs.remove(lsnr);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void setFailState(int failState) {
    if(failState == Fallible.DEAD)
      log("node "+this.getPastryId()+" has been set as DEAD");
    if(failState == Fallible.DOWN)
      log("node "+this.getPastryId()+" has been set as DOWN");
    if(failState == Fallible.OK)
      log("node "+this.getPastryId()+" has been set as OK");
    
    super.setFailState(failState);
  }

  /**
   * Forces this node as in state {@link PastryProtocolState#JOINED}.
   */
  public void setJoined() {
    getPastryProtocol().setJoined();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public String toString() {
    return this.getPastryProtocol().getState()+" "+this.getPastryId().toString();
  }

  /**
   * Returns a {@link String} representation of the local-store.
   * @return A {@link String} representation of the local-store
   */
  public String localStoreAsString() {
    if(localStore!=null)
      return localStore.toString();
    else
      return "[]";
  }

  /**
   * Logs the provided message to the configured log file.
   *
   * @param msg The message
   */
  protected void log(String msg) {
    if(doLog)
      stream.append(CommonState.getTime()+") "+msg+"\n");
  }

  /**
   * Adds the given resource to the local-store.
   *
   * @param resource The resource to add
   */
  private void addResourceToLocalStore(PastryResource resource) {
    if(localStore==null)
      localStore = new HashMap<PastryId, PastryResource>();
    localStore.put(resource.getResourceId(),resource);
  }

  /**
   * Notifies a join event to registered listeners.
   * @param info The join event
   */
  private void notifyJoinListeners(JoinedInfo info) {
    if(joinLsnrs!=null) {
      for(PastryJoinLsnrIfc lsnr : joinLsnrs) {
        lsnr.joined(info);
      }
    }
  }

  /**
   * Notifies a resource assinged event to registered listeners.
   * @param info The event
   */
  private void notifyResourceAssignListeners(ResourceAssignedInfo info) {
    if(resAssignLsnrs!=null) {
      for(PastryResourceAssignLsnrIfc lsnr : resAssignLsnrs) {
        lsnr.resourceAssigned(info);
      }
    }
  }

  /**
   * Notifies a resource discovered event to registered listeners.
   * @param info The event
   */
  private void notifyResourceLookupListeners(ResourceDiscoveredInfo info) {
    if(resLookupLsnrs!=null) {
      for(PastryResourceDiscoveryLsnrIfc lsnr : resLookupLsnrs) {
        lsnr.resourceDiscovered(info);
      }
    }
  }

  /**
   * Notifies a resource received event to registered listeners.
   * @param info The event
   */
  private void notifyResourceReceivedListeners(ResourceReceivedInfo info) {
    if(resRcvdLsnrs!=null) {
      for(PastryResourceAssignLsnrIfc lsnr : resRcvdLsnrs) {
        lsnr.resourceReceived(info);
      }
    }
  }

  /**
   * Notifies a resource routed event to registered listeners.
   * @param info The event
   */
  private void notifyRoutedResourceListeners(ResourceRoutedInfo info) {
    if(resRoutedLsnrs!=null) {
      for(PastryResourceAssignLsnrIfc lsnr : resRoutedLsnrs) {
        lsnr.resourceRouted(info);
      }
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void joined(JoinedInfo info) {
    log("node "+this.getPastryId()+" has joined");
    notifyJoinListeners(info);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void resourceAssigned(ResourceAssignedInfo info) {
    notifyResourceAssignListeners(info);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void resourceDiscovered(ResourceDiscoveredInfo info) {
    log("node "+this.getPastryId()+" has found resource "+info.getResourceId());
    addResourceToLocalStore(info.getResource());
    notifyResourceLookupListeners(info);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void resourceReceived(ResourceReceivedInfo info) {
    if(!localStore.containsKey(info.getResource().getResourceId()))
      log("node "+this.getPastryId()+" has stored resource "+info.getResource());
    addResourceToLocalStore(info.getResource());
    notifyResourceReceivedListeners(info);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void resourceRouted(ResourceRoutedInfo info) {
    log("node "+this.getPastryId()+" has routed resource "+info.getResource());
    notifyRoutedResourceListeners(info);
  }

  private class DelayedLookup {
    private PastryId id;
    private int retries;
    private long resubmitTime;

    @Override
    public String toString() {
      return id+" "+retries;
    }

    private DelayedLookup(PastryId id) {
      this.id = id;
      retries = 0;
      this.resubmitTime = CommonState.getTime()+delayedLookupTime;
    }
  }
}